package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/go-redis/redis/v8"
	_ "github.com/go-sql-driver/mysql"
	"math/rand"
	"strconv"
	"time"
)

type DBConnect struct {
	DBUrl string
	DB    *sql.DB
	//Data  HistoryData
}

type HistoryData struct {
	GatewayID  string     `json:"Gateway_id"`
	DeviceData DeviceData `json:"data"`
}

type DeviceData struct {
	DeviceType uint8 `json:"type"`
	Temp float32 `json:"temp"`
	TempSet float32 `json:"tempset"`
	Speed uint `json:"speed"`
	SpeedSet uint `json:"speedset"`
	HourSet uint8 `json:"Hset"`
	MinSet uint8 `json:"Mset"`
	Hour uint8 `json:"Hour"`
	Min uint8 `json:"Min"`
	TempMin float32 `json:"tempmin"`
	TempMax float32 `json:"tempmax"`
	SpeedMin uint `json:"spmin"`
	SpeedMax uint `json:"spmax"`
	ErrorInfo uint16 `json:"errinfo"`
	State uint16 `json:"state"`
	Mode uint8 `json:"mode"`
	Memory uint8 `json:"memory"`
}
type AlarmData struct {
	GatewayID string
	DeviceType uint8
	Alarm Alarm
}
type Alarm struct {
	InnerSensorErr              uint `json:"inner_sensor_err"`
	InnerSensorTempAlarm        uint `json:"inner_sensor_temp_alarm"`
	ExternalSensorErr           uint `json:"external_sensor_err"`
	ExternalSensorOverTemp      uint `json:"external_sensor_over_temp"`
	MechanicalLimitTempOverTemp uint `json:"mechanical_limit_temp_over_temp"`
	MechanicalLimitSensorErr    uint `json:"mechanical_limit_sensor_err"`
	MotorStalling               uint `json:"motor_stalling"`
	ExternalSensorDrop          uint `json:"external_sensor_drop"`
	EndOfTime                   uint `json:"end_of_time"`
	ExternalSensorNotPlugged    uint `json:"external_sensor_not_plugged"`
}
type SqlConfig struct {
	Username string
	Password string
	Host string
	Port string
	Database string
}

type MqttConfig struct {
	broker string
	clientID string
}
var ctx = context.Background()
// 获取消息间隔,设备每秒发送至mqtt服务器,mqtt服务器每五秒采样一次
var duration int
var redisConfig = redis.Options{
	Network:   "tcp",
	Addr:      "localhost:6379",
	Dialer:    nil,
	OnConnect: onRedisConnect,
	DB:        0,
}
var sqlconfig = SqlConfig{
	Username: "root",
	Password: "123456",
	Host:     "localhost",
	Port:     "3306",
	Database: "test",
}
var mqttconfig = MqttConfig{
	broker:     "tcp://orderer1.dc.dc:1883",
	clientID: "history-data-client" + strconv.Itoa(rand.Intn(10000)),
}
func newSqlConn() *DBConnect {
	dburl := sqlconfig.Username+":"+sqlconfig.Password+"@tcp("+sqlconfig.Host+":"+sqlconfig.Port+")/"+sqlconfig.Database+"?charset=utf8mb4"
	return &DBConnect{
		//DBUrl: "root:123456@tcp(localhost:3306)/test?charset=utf8mb4",
		DBUrl:dburl,
	}
}

func onRedisConnect(ctx context.Context, cn *redis.Conn) error {
	fmt.Println("connected")
	return nil
}

func onMqttConnect(c mqtt.Client) {
	fmt.Println("connected")
}
func msgHandler(c mqtt.Client, msg mqtt.Message) {
	var err error
	db := newSqlConn()
	db.DB, err = sql.Open("mysql", db.DBUrl)
	if err != nil {
		fmt.Printf("sql connect error : %s\n", err)
	}
	err = db.DB.Ping()
	if err != nil {
		fmt.Println(err.Error())
	} else {
		fmt.Println("connected ok")
	}

	//
	var hd HistoryData
	var ad AlarmData
	if duration == 0 {
		var errorInfo uint16
		fmt.Println(msg.Topic())
		fmt.Println(string(msg.Payload()))

		err :=  json.Unmarshal(msg.Payload(), &hd)
		if err != nil {
			panic(err)
		}
		
		errorInfo = hd.DeviceData.ErrorInfo

		a := getDeviceAlarm(errorInfo)
		ad = AlarmData{
			GatewayID:  hd.GatewayID,
			DeviceType: hd.DeviceData.DeviceType,
			Alarm:      a,
		}
		db.insertAlarmData(&ad)
		db.insertHistoryData(&hd)
		duration = 4
	} else {
		db.insertAlarmData(&ad)
		duration--
	}
}
func getDeviceAlarm(e uint16) Alarm {
	return Alarm{
		InnerSensorErr:              uint(e & 32768),   //0
		InnerSensorTempAlarm:        uint(e & 16384),   //1
		ExternalSensorErr:           uint(e & 8192),    //2
		ExternalSensorOverTemp:      uint(e & 4096),
		MechanicalLimitTempOverTemp: uint(e & 2048),
		MechanicalLimitSensorErr:    uint(e & 1024),
		MotorStalling:               uint(e & 512),     //6
		ExternalSensorDrop:          uint(e & 256),
		EndOfTime:                   uint(e & 128),
		ExternalSensorNotPlugged:    uint(e & 64),      //9
	}
}
func mqttPrecess(topic string) {
	clientOption := mqtt.NewClientOptions().AddBroker(mqttconfig.broker).SetClientID(mqttconfig.clientID)
	clientOption.OnConnect = onMqttConnect
	clientOption.SetDefaultPublishHandler(msgHandler)
	client := mqtt.NewClient(clientOption)

	if token := client.Connect(); token.Wait() && token.Error() != nil {
		panic(token.Error())
	}
	if token := client.Subscribe(topic, 0, nil); token.Wait() && token.Error() != nil {
		fmt.Println("test sub")
	}

}
func redisProcess() {
	c := redis.NewClient(&redisConfig)
	for {
		res, err := c.LPop(ctx, "mqtt_subscribe").Result()
		if err != nil {
			fmt.Println(err.Error())
		}
		if res != "" {
			topic := res
			go mqttPrecess(topic)
		} else {
			time.Sleep(2 * time.Second)
		}
	}
}
func (dbconn *DBConnect) insertHistoryData(hd *HistoryData) {
	//var err error
	//dbconn.DB, err = sql.Open("mysql", dbconn.DBUrl)
	//defer dbconn.DB.Close()
	//if err != nil {
	//	fmt.Println("sql connect error : %s", err)
	//}
	//err = dbconn.DB.Ping()
	//if err != nil {
	//	fmt.Println(err.Error())
	//} else {
	//	fmt.Println("connected ok")
	//}
	//res, err := dbconn.DB.Exec("create table if not exists student (id int auto_increment primary key , stuname varchar (15) not null default '');")
	//fmt.Println(res.RowsAffected())
	dbprepare, err := dbconn.DB.Prepare("insert into hh_history_data (gateway_id, upload_time, device_type, temp, speed, display_time, status, mode, lossElecMem) values(?,?,?,?,?,?,?,?,?);")
	displayTime := strconv.FormatUint(uint64(hd.DeviceData.Hour),10) + ":" + strconv.FormatUint(uint64(hd.DeviceData.Min),10)
	result, err := dbprepare.Exec(hd.GatewayID, time.Now().String(),hd.DeviceData.DeviceType, hd.DeviceData.Temp, hd.DeviceData.Speed, displayTime, hd.DeviceData.State, hd.DeviceData.Mode, hd.DeviceData.Memory)
	fmt.Println(result)
	if err != nil {
		panic(err)
	}
}

func (dbconn *DBConnect) insertAlarmData(alarmData *AlarmData) {
	//var err error
	//dbconn.DB, err = sql.Open("mysql", dbconn.DBUrl)
	//defer dbconn.DB.Close()
	//if err != nil {
	//	fmt.Printf("sql connect error : %s\n", err)
	//}
	//err = dbconn.DB.Ping()
	//if err != nil {
	//	fmt.Println(err.Error())
	//} else {
	//	fmt.Println("connected ok")
	//}
	//res, err := dbconn.DB.Exec("create table if not exists student (id int auto_increment primary key , stuname varchar (15) not null default '');")
	//fmt.Println(res.RowsAffected())
	dbprepare, err := dbconn.DB.Prepare("insert into hh_device_alarm (gateway_id, device_type, alarm_date, inner_sensor_err, inner_sensor_temp_alarm, external_sensor_err, external_sensor_over_temp, mechanical_limit_temp_over_temp, mechanical_limit_sensor_err, motor_stalling, external_sensor_drop, end_of_time, external_sensor_not_plugged) values(?,?,?,?,?,?,?,?,?,?,?,?,?);")
	//displayTime := strconv.FormatUint(uint64(hd.DeviceData.Hour),10) + ":" + strconv.FormatUint(uint64(hd.DeviceData.Min),10)
	result, err := dbprepare.Exec(alarmData.GatewayID, alarmData.DeviceType, time.Now().String(), alarmData.Alarm.InnerSensorErr, alarmData.Alarm.InnerSensorTempAlarm, alarmData.Alarm.ExternalSensorErr, alarmData.Alarm.ExternalSensorOverTemp, alarmData.Alarm.MechanicalLimitTempOverTemp, alarmData.Alarm.MechanicalLimitSensorErr, alarmData.Alarm.MotorStalling, alarmData.Alarm.ExternalSensorDrop, alarmData.Alarm.EndOfTime, alarmData.Alarm.ExternalSensorNotPlugged)
	fmt.Println(result)
	if err != nil {
		panic(err)
	}
}
func main() {
	go redisProcess()

	select {}
}
